package com.zzx.graduation.service;


import com.alibaba.fastjson.JSONObject;
import com.zzx.graduation.entity.KafkaNews;
import com.zzx.graduation.entity.News;
import com.zzx.graduation.entity.User;
import com.zzx.graduation.repository.GroupRepository;
import com.zzx.graduation.repository.NewsRepository;
import com.zzx.graduation.repository.UserRepository;
import com.zzx.graduation.utils.KafkaBatchConsumer;
import com.zzx.graduation.utils.KafkaSender;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;


@Component
public class NewsService {

    @Autowired
    private UserRepository userRepository;
    @Autowired
    private GroupRepository groupRepository;
    @Autowired
    private NewsRepository newsRepository;

        @Autowired
        KafkaSender sender;

//    @Test
//    public void send() {
//        sender.send("这是Kafka发送的消息内容" + System.currentTimeMillis());
//
//    }

    public List<News> findNewsBySourceId(Integer id){
        List<News> allBySourceId = newsRepository.findAllBySourceId(id);
        return allBySourceId;
    }
    public List<News> findNewByTargetIdAndType(Integer id, Integer type){
        List<News> news = newsRepository.findAllByTargetIdAndTargetType(id, type);
        return news;
    }


    /**
     * 发送消息的只有员工，
     *      * 当消息发往群聊的时候，消息要中转到kafka，文件存入服务器
     *      * 消息主题按照群聊，广播到所有人
     *      *
     * @param news
     * @return
     */
    public News addNews(News news) {
        //建立关联关系
        Integer sourceId = news.getSourceId();
        User user = userRepository.getOne(sourceId);
        //记得注释掉
       // user.setStaffAddress(new Date().toString());
        news.setOwnUser(user);
        News newsSave = newsRepository.save(news);
        System.out.println(newsSave);

        Integer targetId = news.getTargetId();
        Integer targetType = news.getTargetType();

        //如果消息发往群聊，则消息要发到kafka,主题tpoic为：group--targetid(群id)
        if (targetType == 1) {
            String s = JSONObject.toJSONString(toKafkaNews(news));
            sender.send("group--" + targetId, s);


//            new Thread(new Runnable() {
//                @Override
//                public void run() {
//                    //测试消费
//                    KafkaBatchConsumer kafkaBatchConsumer = new KafkaBatchConsumer("group--" + targetId,"1") {
//                        @Override
//                        public void hanldeRecord(ConsumerRecord<String, String> record) {
//                            System.out.println("消费--->" + record.value() );
//                        }
//                    };
//                    kafkaBatchConsumer.receivebatch();
//                }
//            }).start();


        }else {
            //否则。消息发往个人主题 topic:  id(小)--id（大）
//           String topic = sourceId < targetId ? sourceId + "--" + targetId（个人id）
//                   : targetId + "--" + sourceId;
            //发往个人主题邮箱topic： mail--id （代表这个id的人的收件箱，存放收到的消息）
            String topic = "mail--" + targetId;
           // System.out.println("xiaoxi >>>>>1111" );
           String s = JSONObject.toJSONString(toKafkaNews(news));
            System.out.println("发送数据到kafka：" + s);
            //System.out.println("xiaoxi >>>>>" );
            sender.send(topic,s);

//           new Thread(new Runnable() {
//               @Override
//               public void run() {
//                   //测试消费
//                   KafkaBatchConsumer kafkaBatchConsumer = new KafkaBatchConsumer("mail--" + targetId,"1") {
//                       @Override
//                       public void hanldeRecord(ConsumerRecord<String, String> record) {
//                           System.out.println("消费--->" + record.value() );
//                       }
//                   };
//                   kafkaBatchConsumer.receivebatch();
//               }
//           }).start();

        }


        return newsSave;
    }

    public KafkaNews toKafkaNews(News news) {
        KafkaNews kafkaNews = new KafkaNews();
        kafkaNews.setAgreement(news.getAgreement());
        kafkaNews.setContent(news.getContent());
        kafkaNews.setId(news.getId());
        kafkaNews.setSourceId(news.getSourceId());
        kafkaNews.setTargetId(news.getTargetId());
        kafkaNews.setTargetType(news.getTargetType());
        kafkaNews.setTime(news.getTime());
        return kafkaNews;
    }
    //一般不用
    public Boolean deleteNews(Integer id){
        newsRepository.deleteById(id);
        return true;
    }

    //一般不用
    public News modifyNews(News news) {
        News save = newsRepository.save(news);
        return save;
    }
}
